package com.qisi.demo;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkCDC_DS {
    // TODO 测试：mysql 操作：增加 修改 删除
    //  insert into gmall1116_realtime.user_test values('10009','yang9');
    //  update gmall1116_realtime.user_test set username='yang10' where userid='10009';
    //  delete from  gmall1116_realtime.user_test where userid='10009';


    public static void main(String[] args) throws Exception {
        // TODO --用 createLocalEnvironmentWithWebUI 可以在windows上运行访问 webUI页面 localhost:8081 访问提示找不到ip
        //   http://192.168.253.1:8081/
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(Integer.parseInt(args[0])); //因不是从kafka读数据所以这里用1个并行度，是直接读mysql的binlog数据
        env.setParallelism(1);
        // TODO --设置检查点:1启动4设置
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall/cdc"));
//        env.setStateBackend(new FsStateBackend("file:///qisi/gitlab/flink-real-time-alert/cp/gmall/cdc"));
        System.setProperty("HADOOP_USER_NAME","atguigu");// windows 上操作hadoop需设置操作hadoop的用户 否则没有权限，linux不用设置

        DebeziumSourceFunction<String> debeziumSourceFunction = MySQLSource
                .<String>builder() // TODO --前面 <String> 为指定的 构建数据流中的类型，必须写，否则会报错
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("gmall1116")
                .tableList("gmall1116.user_info")// 表名前必须加数据库名、否则监听不到表的初始数据
                .startupOptions(StartupOptions.initial())
                //.deserializer(new StringDebeziumDeserializationSchema())  // 反序列化后的格式不太友好：封装的Source..对象
                .deserializer(new StringDebeziumDeserializationSchema()) // 可以自定义采集数据的反序列化方式
                .build();// 构建者模式创建对象  --构建前可以先对属性赋值
            // {"database":"gmall1116_realtime","data":{"userid":"10009","username":"yang10"},"type":"delete","table":"user_test"}
        // TODO StartupOptions启动选项：程序启动时对数据的获取方式 ： initial启动程序时是从表中获取数据 后续再从binlog中获取变化数据、其他启动时是从binlog中获取数据（binlog中可能不包含历史数据）。
        //  initial 每次启动程序时都执行快照：监听数据库/表的（是从表中去查数据），读取到表中的全部数据后，后序直接取binlog中读取变化的数据
        //  earliest 程序启动时是从binlog的最早一条数据开始读取
        //  latest  程序启动时是从最后一条binlog文件读取
        //  specificOffset（具体文件、具体偏移量位置） --程序启动时直接从指定的偏移量位置开始读binlog文件
        //  timestamp(时间搓) 程序启动时只读取binlog中大于指定时间搓的数据
        env.addSource(debeziumSourceFunction).print();
        env.execute();
    }
}